package com.jzo2o.foundations.listener;

import com.jzo2o.canal.listeners.AbstractCanalRabbitMqMsgListener;
import com.jzo2o.common.expcetions.CommonException;
import com.jzo2o.es.core.ElasticSearchTemplate;
import com.jzo2o.foundations.model.domain.ServeSync;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Component
public class DataSynchronismListener extends AbstractCanalRabbitMqMsgListener<ServeSync> {
    @Resource
    private ElasticSearchTemplate elasticSearchTemplate;


    @RabbitListener(queues = "canal-mq-jzo2o-foundations")
    public void dataSynchronism(Message message) throws Exception {
        parseMsg(message);
    }

    /**
     * 重写AbstractCanalRabbitMqMsgListener中的添加方法
     * 实现Mysql和ES的数据同步
     * @param data 数据
     */
    @Override
    public void batchSave(List<ServeSync> data) {
        Boolean isTrue = elasticSearchTemplate.opsForDoc().batchUpsert("serve_aggregation", data);
        if(!isTrue) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new CommonException("同步失败");
        }
    }

    /**
     * 重写AbstractCanalRabbitMqMsgListener中的删除方法
     * 实现Mysql和ES的数据同步
     * @param ids ids
     */
    @Override
    public void batchDelete(List<Long> ids) {
        Boolean isTrue = elasticSearchTemplate.opsForDoc().batchDelete("serve_aggregation", ids);
        if(!isTrue) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new CommonException("同步失败");
        }
    }
}
